#!/bin/bash
# set -x
KAFKA_HOME=/disk2/kafka_2.10-0.8.2.2
while true; do
	# for raw-binary input
        exception_processed_offset_total=0
        exception_latest_offset_total=0
        time=$(date +%s)
        for i in $(cat exception_offsets)
        do
          exception_processed_offset=$(echo "get /zk_test/dataplatform/storm/hdfs-kafka-spout_anomaly/$i" | $KAFKA_HOME/bin/zookeeper-shell.sh kmxnode6,kmxnode4,kmxnode3  2>/dev/null | tail -n 1 | jq '.offset')
          exception_processed_offset_total=$(expr $exception_processed_offset_total + $exception_processed_offset)
        done
        for h in $($KAFKA_HOME/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.1.88:19092,192.168.1.89:19092 -time -1 --topic defaultUser-defaultSpace-exception | awk -F [:] '{print $3}')
        do
          exception_latest_offset_total=$(expr $exception_latest_offset_total + $h)
        done
	echo "put sdlg.kafka.exception.offset.processed.total $time $exception_processed_offset_total host=$(hostname -s)"
        echo "put sdlg.kafka.exception.offset.latest.total $time $exception_latest_offset_total host=$(hostname -s)"
	sleep 60
done #| nc 211.154.7.134 4242
